#include "../include/output.h"

logSink::logSink()
{
}

logSink::~logSink()
{
}

void logOutputStdout::logOut(const std::string& log)
{
	std::cout << log;
}

logOutputFile::logOutputFile(const std::string& path)
{
	_file = fopen(path.c_str(), "w+");
	if (_file == nullptr)
	{
		exit(-1);
	}
}

void logOutputFile::logOut(const std::string& log)
{
	fwrite(log.c_str(), 1, log.size(), _file);
}

logOutputFile::~logOutputFile()
{
	fclose(_file);
	_file = nullptr;
}

void logger::formatChange(const std::string& key, const std::function<std::string(logMsg)>& fun)
{
	_reformat.change(key, fun);
}

void logger::formatClear()
{
	_reformat.clear();
}

void logger::setPattern(const std::string& logFormat)
{
	_reformat.setPattern(logFormat);
}

void synchronizeLogger::operator()(const std::string& data, int level, const std::string& fileName, size_t line)
{
	logMsg log(data, level, _loggerName, fileName, line);

	{
		std::unique_lock<std::mutex> lock(_mutex);
		for (auto& sink : _logSink)
		{
			sink->logOut(_reformat.format(log));
		}
	}
}

msgBuffer::msgBuffer(const Fun& outputCall, size_t maxCount)
	: _outputCall(outputCall)
	, _maxCount(maxCount)
	, _check(true)
	, _buffer1(maxCount)
	, _buffer2(maxCount)
	, _pushIt(_buffer1.begin())
	, _getIt(_buffer2.begin())
	, _loop(&msgBuffer::execute, this)
{
}

void msgBuffer::stop()
{
	_check = false;
	_conGet.notify_all();
	_loop.join();
}

void msgBuffer::push(logMsg&& msg)
{
	if (!_check)
	{
		return;
	}
	{
		std::unique_lock<std::mutex> lock(_mutex);
		_conPush.wait(lock, [&]() { return _pushIt != _buffer1.end(); });
		(*_pushIt) = std::move(msg);
		_pushIt++;
	}
	_conGet.notify_all();
}

msgBuffer::~msgBuffer()
{
	stop();
}

void msgBuffer::execute()
{
	while (true)
	{
		{
			std::unique_lock<std::mutex> lock(_mutex);
			if (!_check && _buffer1.begin() == _pushIt)
			{
				return;
			}

			_conGet.wait(lock, [&]() { return _buffer1.begin() != _pushIt || !_check; });
			_buffer1.swap(_buffer2);
			_getIt = _pushIt;
			_pushIt = _buffer1.begin();
		}
		_conPush.notify_all();
		_outputCall(_buffer2, _getIt);
		_getIt = _buffer2.begin();
	}
}

void asynchronousLogger::operator()(const std::string& data, int level, const std::string& fileName, size_t line)
{
	_buffer.push(logMsg(data, level, _loggerName, fileName, line));
}

void asynchronousLogger::outputCall(std::vector<logMsg>& t1, std::vector<logMsg>::iterator& t2)
{
	if (_logSink.empty())
	{
		return;
	}
	for (auto& sink : _logSink)
	{
		for (auto it = t1.begin(); it != t2; it++)
		{
			sink->logOut(_reformat.format(*it));
		}
	}
}

void asyDisruptorLogger::operator()(const std::string& data, int level, const std::string& fileName, size_t line)
{
	//申请序列位置
	int64_t sequence = _sequencer.Claim();

	//写入数据
	_sequencer[sequence].insert(data, level, _loggerName, fileName, line);

	//发布事件
	_sequencer.Publish(sequence);
}

asyDisruptorLogger::~asyDisruptorLogger()
{
	if (_conBarrier)
	{
		_conBarrier->set_alerted(true);
		int64_t sequence = _sequencer.Claim();
		_sequencer[sequence].insert("", static_cast<int>(Level::OFF), "-1", "", 0); // 空事件
		_sequencer.Publish(sequence);
	}

	if (_thread.joinable())
	{
		_thread.join();
	}
}

void asyDisruptorLogger::execute()
{
	//起始序列
	int64_t nextSequence = disruptor::kFirstSequenceValue;

	while (true)
	{
		//等待事件
		int64_t triggerSequence = _conBarrier->WaitFor(nextSequence);

		if (_conBarrier->alerted())
		{
			if (triggerSequence >= nextSequence)
			{
				for (auto& sink : _logSink)
				{
					for (int64_t seq = nextSequence; seq <= triggerSequence; seq++)
					{
						if (_sequencer[seq]._loggerName == "-1")
						{
							return;
						}
						sink->logOut(_reformat.format(_sequencer[seq]));
					}
				}
			}
			break; 
		}
		if (triggerSequence >= nextSequence)  
		{
			for (auto& sink : _logSink)
			{
				for (int64_t seq = nextSequence; seq <= triggerSequence; seq++)
				{
					if (_sequencer[seq]._loggerName == "-1")
					{
						return;
					}
					sink->logOut(_reformat.format(_sequencer[seq]));
				}
			}
			nextSequence = triggerSequence + 1;  
		}

		_conSeq.set_sequence(triggerSequence); //通知生产者复用
	}
}

logOutputMmapFile::logOutputMmapFile(const std::string& path)
	: _file(path)
{

}

void logOutputMmapFile::logOut(const std::string& log)
{
	_file.write(log);
}
